package com.example.springbootrocketmq.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.UnsupportedEncodingException;

/**
 * MQ顺序消费，消费者
 * 会发现order  的id为2的订单总是落到队列2中去
 */
public class OrderConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("HAOKE_ORDER_CONSUMER");
        consumer.setNamesrvAddr("192.168.64.128:9876");
        consumer.subscribe("haoke_order_topic", "*");
        consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                try {
                    System.out.println(Thread.currentThread().getName() + " " + msg.getQueueId() + " " + new String(msg.getBody(),"UTF-8"));
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
//                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
            return ConsumeOrderlyStatus.SUCCESS;
        });
        consumer.start();
    }
}